#include "tsdb.h"
//#include "zmalloc.h"
#include <stdlib.h>
#ifndef C_OK
#define C_OK	(0)
#endif // !C_OK

#ifndef C_ERR
#define C_ERR	(-1)
#endif // !C_ERR

/* -------------------------- static variables ------------------------------- */

static const uint32_t kLeadingZerosLengthBits = 5;
static const uint32_t kBlockSizeLengthBits = 6;
static const uint32_t kMaxLeadingZerosLength = (1 << 5) - 1;
static const uint32_t kBlockSizeAdjustment = 1;
static const uint32_t kDefaultDelta = 60;
static const uint32_t kBitsForFirstTimestamp = 31; // Works until 2038.

static struct {
	int64_t bitsForValue;
	uint32_t controlValue;
	uint32_t controlValueBitLength;
} const timestampEncodings[4] = {
	{ 7, 2, 2 },
	{ 9, 6, 3 },
	{ 12, 14, 4 },
	{ 32, 15, 4 } };

/* -------------------------- private prototypes ---------------------------- */
static int tstreamInit(tstream *t);
// Compression methods.
static int tstreamAppendTimestamp(tstream *t, int64_t timestamp, int64_t minTimestampDelta);
static void tstreamAppendValue(tstream *t, double value);

// sds read write in bits methods
// Adds a value to a sds string. `bitsInValue` specifies the number
// of least significant bits that will be added to the bit
// string. The bits from `value` will be added from the most
// significant bit to the least significant bit.
static sds sdscatbits(sds s, uint64_t value, uint64_t bitsInValue, uint32_t *numBits);
// Reads a value from a sds string. `bitPos` is updated by
// `bitsToRead`. `bitsToRead` must be 64 or less.
static uint64_t sdsreadbits(sds s, uint64_t *bitPos, uint32_t bitsToRead);
// Finds the first zero bit and returns its distance from bitPos. If
// not found within limit, returns limit.
static uint32_t sdsfirstzero(sds s, uint64_t *bitPos, uint32_t limit);

static int64_t sdsReadNextTimestamp(sds s, uint64_t *bitPos, int64_t *prevTimestampValue, int64_t *prevTimestampDelta);

static double sdsReadNextValue(sds s, uint64_t *bitPos, uint64_t *previousValue, uint8_t* previousLeadingZeros, uint8_t *previousTrailingZeros);
/* ----------------------------- API implementation ------------------------- */
tstream *tstreamCreate(void) {
	tstream *t = t_malloc(sizeof(*t));
	tstreamInit(t);
	return t;
}

int tstreamInit(tstream *t) {
	t->data = NULL;
	tstreamReset(t);
  return C_OK;
}

void tstreamReset(tstream *t) {
	t->first_timestamp = 0;
	t->prev_timestamp = 0;
	t->num_values = 0;
	t->num_bits = 0;
	t->prev_timestamp_delta = 0;
	t->prev_value = 0;
	t->prev_value_leading_zeros = 0;
	t->prev_value_trailing_zeros = 0;
	if (t->data)
        sdsfree(t->data);
	t->data = sdsempty();
}

void tstreamRelease(tstream *t) {
    sdsfree(t->data);
    t_free(t);
}

int tstreamAppendtv(tstream *t, tvalue tv, int64_t minTimestampDelta) {
	return tstreamAppend(t, tv.timestamp, tv.value, minTimestampDelta);
}

int tstreamAppend(tstream *t, int64_t timestamp, double value, int64_t minTimestampDelta) {
	if (tstreamAppendTimestamp(t, timestamp, minTimestampDelta) != C_OK) {
		return C_ERR;
	}

	tstreamAppendValue(t, value);
	t->num_values++;
	return C_OK;
}

int tstreamAppendTimestamp(tstream *t, int64_t timestamp, int64_t minTimestampDelta) {
	// Store a delta of delta for the rest of the values in one of the
	// following ways
	//
	// '0' = delta of delta did not change
	// '10' followed by a value length of 7
	// '110' followed by a value length of 9
	// '1110' followed by a value length of 12
	// '1111' followed by a value length of 32

	int64_t delta = timestamp - t->prev_timestamp;
	// Skip the minTimestampDelta check for the first timestamp.
	if (delta < minTimestampDelta && t->prev_timestamp != 0) {
		return C_ERR;
	}


	if (sdslen(t->data) == 0) {
		// Store the first value as is
		t->data = sdscatbits(t->data, timestamp, kBitsForFirstTimestamp, &t->num_bits);
		t->first_timestamp = timestamp;
		t->prev_timestamp = timestamp;
		t->prev_timestamp_delta = kDefaultDelta;
		return C_OK;
	}

	int64_t deltaOfDelta = delta - t->prev_timestamp_delta;

	if (deltaOfDelta == 0) {
		t->prev_timestamp = timestamp;
		t->data = sdscatbits(t->data, 0, 1, &t->num_bits);
		return C_OK;
	}

	if (deltaOfDelta > 0) {
		// There are no zeros. Shift by one to fit in x number of bits
		deltaOfDelta--;
	}

	int64_t absValue = llabs(deltaOfDelta);

	for (int i = 0; i < 4; i++) {
		if (absValue < ((int64_t)1 << (timestampEncodings[i].bitsForValue - 1))) {
			t->data = sdscatbits(t->data,
				timestampEncodings[i].controlValue,
				timestampEncodings[i].controlValueBitLength,
				&t->num_bits);

			// Make this value between [0, 2^timestampEncodings[i].bitsForValue - 1]
			int64_t encodedValue = deltaOfDelta +
				((int64_t)1 << (timestampEncodings[i].bitsForValue - 1));
			t->data = sdscatbits(t->data,
				encodedValue,
				timestampEncodings[i].bitsForValue,
				&t->num_bits);
			break;
		}
	}

	t->prev_timestamp = timestamp;
	t->prev_timestamp_delta = delta;

	return C_OK;
}

void tstreamAppendValue(tstream *t, double value) {
	uint64_t* p = (uint64_t*)&value;
	uint64_t xorWithPrevius = t->prev_value ^ *p;

	// Doubles are encoded by XORing them with the previous value.  If
	// XORing results in a zero value (value is the same as the previous
	// value), only a single zero bit is stored, otherwise 1 bit is
	// stored. TODO : improve this with RLE for the number of zeros
	//
	// For non-zero XORred results, there are two choices:
	//
	// 1) If the block of meaningful bits falls in between the block of
	//    previous meaningful bits, i.e., there are at least as many
	//    leading zeros and as many trailing zeros as with the previous
	//    value, use that information for the block position and just
	//    store the XORred value.
	//
	// 2) Length of the number of leading zeros is stored in the next 5
	//    bits, then length of the XORred value is stored in the next 6
	//    bits and finally the XORred value is stored.
	if (xorWithPrevius == 0) {
		t->data = sdscatbits(t->data, 0, 1, &t->num_bits);
		return;
	}

	t->data = sdscatbits(t->data, 1, 1, &t->num_bits);

	uint32_t leadingZeros = __builtin_clzll(xorWithPrevius);
	uint32_t trailingZeros = __builtin_ctzll(xorWithPrevius);

	if (leadingZeros > kMaxLeadingZerosLength) {
		leadingZeros = kMaxLeadingZerosLength;
	}

	int blockSize = 64 - leadingZeros - trailingZeros;
	uint32_t expectedSize =
		kLeadingZerosLengthBits + kBlockSizeLengthBits + blockSize;
	uint32_t previousBlockInformationSize =
		64 - t->prev_value_trailing_zeros - t->prev_value_leading_zeros;

	if (leadingZeros >= t->prev_value_leading_zeros &&
		trailingZeros >= t->prev_value_trailing_zeros &&
		previousBlockInformationSize < expectedSize) {
		// Control bit for using previous block information.
		t->data = sdscatbits(t->data, 1, 1, &t->num_bits);

		uint64_t blockValue = xorWithPrevius >> t->prev_value_trailing_zeros;
		t->data = sdscatbits(t->data, blockValue, previousBlockInformationSize, &t->num_bits);
	} else {
		// Control bit for not using previous block information.
		t->data = sdscatbits(t->data, 0, 1, &t->num_bits);

		t->data = sdscatbits(t->data, leadingZeros, kLeadingZerosLengthBits, &t->num_bits);

		// To fit in 6 bits. There will never be a zero size block
		t->data = sdscatbits(
			t->data,
			 blockSize - kBlockSizeAdjustment,
			kBlockSizeLengthBits,
			&t->num_bits);

		uint64_t blockValue = xorWithPrevius >> trailingZeros;
		t->data = sdscatbits(t->data, blockValue, blockSize, &t->num_bits);

		t->prev_value_trailing_zeros = trailingZeros;
		t->prev_value_leading_zeros = leadingZeros;
	}

	t->prev_value = *p;
}

int64_t tstreamGetFirstTimeStamp(tstream *t) {
	if (!t || !t->data) return 0;
	if (sdslen(t->data) == 0) return 0;
	uint64_t bitPos = 0;
	if (t->first_timestamp == 0) {
		t->first_timestamp = sdsreadbits(t->data, &bitPos, kBitsForFirstTimestamp);
	}
	return t->first_timestamp;
}

int64_t tstreamGetPreviousTimeStamp(tstream *t) {
	if (!t) return 0;
	return t->prev_timestamp;
}

/* ----------------------- iterator api -------------------------------------*/
void tstreamInitIterator(tsiter *iter){
	iter->prev_timestamp = 0;
	iter->prev_value = 0;
	iter->prev_lead_zeros = 0;
	iter->prev_trail_zeros = 0;
	iter->bit_pos = 0;
	iter->prev_timestamp_delta = kDefaultDelta;
}

int tstreamHasNext(tsiter *iter, tstream *t) {
	if (t->num_bits > 0) {
		if (iter->bit_pos >= t->num_bits) return C_ERR;
	} else {
		uint32_t totalBitLen = (sdslen(t->data)-1)*8; /* WARNING: real bits len maybe less than buf bit len*/
		if (iter->bit_pos >= totalBitLen) return C_ERR;
	}

	return C_OK;
}

int tstreamNextValue(tsiter *iter, tstream *t, tvalue *tv) {
	// iterator must init
	if (!iter) return C_ERR;
	// t and sds must init
	if (!t || !t->data) return C_ERR;
	// tv must init
	if (!tv) return C_ERR;

	sds s = t->data;
	if (iter->bit_pos == 0) {
		// read first timestamp
		int64_t firstTimestamp = sdsreadbits(s, &iter->bit_pos, kBitsForFirstTimestamp);
		iter->prev_timestamp = firstTimestamp;
		double firstValue = sdsReadNextValue(s, &iter->bit_pos, &iter->prev_value, &iter->prev_lead_zeros, &iter->prev_trail_zeros);
		tv->timestamp = firstTimestamp;
		tv->value = firstValue;
		return C_OK;
	}


	if (t->num_bits > 0)  {
		if (iter->bit_pos >= t->num_bits) return C_ERR;
	} else {
		// the ts data is read from api. not insert. so num_bits is zero
		uint32_t totalBitLen = (sdslen(t->data)-1)*8; /* WARNING: real bits len maybe less than buf bit len*/
		if (iter->bit_pos >= totalBitLen) return C_ERR;
	}

	int64_t unixTime = sdsReadNextTimestamp(s, &iter->bit_pos, &iter->prev_timestamp, &iter->prev_timestamp_delta);
	double value = sdsReadNextValue(s, &iter->bit_pos, &iter->prev_value, &iter->prev_lead_zeros, &iter->prev_trail_zeros);
	tv->timestamp = unixTime;
	tv->value = value;
	return C_OK;
}

/* Read ts data into stream */
void tstreamReadData(tstream *t, sds s) {
	tstreamReset(t);
	t->data = sdscatsds(t->data, s);
	tsiter iter;
	tstreamInitIterator(&iter);
	size_t num_values = 0;
	if (sdslen(t->data)*8 < kBitsForFirstTimestamp) {
		// invalid sds length
		return;
	}
	// read first_timestamp
	tvalue first;
	tstreamNextValue(&iter, t, &first);
	t->first_timestamp = first.timestamp;
	num_values++;
	while(tstreamHasNext(&iter, t) != C_ERR) {
		tvalue tv;
		tstreamNextValue(&iter, t, &tv);
		num_values++;
	}

	t->prev_timestamp = iter.prev_timestamp;
	t->prev_timestamp_delta = iter.prev_timestamp_delta;
	t->prev_value_leading_zeros = iter.prev_lead_zeros;
	t->prev_value_trailing_zeros = iter.prev_trail_zeros;
	t->prev_value = iter.prev_value;
	t->num_bits = iter.bit_pos;
	t->num_values = num_values;

}

/* ------------------------ sds implements ------------------------------ */
sds sdscatbits(sds s, uint64_t value, uint64_t bitsInValue, uint32_t *numBits) {
	uint32_t bitsAvailable = (*numBits & 0x7) ? (8 - (*numBits & 0x7)) : 0;
	*numBits += bitsInValue;
	if (bitsInValue <= bitsAvailable) {
		// Everything fits inside the last byte
		char apdata = (char)(value << (bitsAvailable - bitsInValue));
		s[sdslen(s) - 1] += apdata;
		return s;
  }

	uint32_t bitsLeft = bitsInValue;
	if (bitsAvailable > 0) {
		// Fill up the last byte
		s[sdslen(s) - 1] += (value >> (bitsInValue - bitsAvailable));
		bitsLeft -= bitsAvailable;
	}

	while (bitsLeft >= 8) {
    // Enough bits for a dedicated byte
    char ch = (value >> (bitsLeft - 8)) & 0xFF;
    s = sdscatlen(s, &ch, 1);
    bitsLeft -= 8;
  }

	if (bitsLeft != 0) {
		// Start a new byte with the rest of the bits
		char ch = (value & ((1 << bitsLeft) - 1)) << (8 - bitsLeft);
		s = sdscatlen(s, &ch, 1);
  }

	return s;
}

uint64_t sdsreadbits(sds s, uint64_t *bitPos, uint32_t bitsToRead) {
	uint64_t value = 0;
  for (uint32_t i = 0; i < bitsToRead; i++) {
    value <<= 1;
    uint64_t bit = (s[*bitPos >> 3] >> (7 - (*bitPos & 0x7))) & 1;
    value += bit;
    (*bitPos)++;
  }

  return value;
}

uint32_t sdsfirstzero(sds s, uint64_t *bitPos, uint32_t limit) {
	uint32_t bits = 0;

  while (bits < limit) {
    uint32_t bit = sdsreadbits(s, bitPos, 1);
    if (bit == 0) {
      return bits;
    }

    bits++;
  }

  return bits;
}

int64_t sdsReadNextTimestamp(sds s, uint64_t *bitPos, int64_t *prevTimestampValue, int64_t *prevTimestampDelta) {
	uint32_t type = sdsfirstzero(s, bitPos, 4);
	if (type > 0) {
    // Delta of delta is non zero. Calculate the new delta. `index`
    // will be used to find the right length for the value that is
    // read.
    int index = type - 1;
    int64_t decodedValue = sdsreadbits(
        s, bitPos, timestampEncodings[index].bitsForValue);

    // [0,255] becomes [-128,127]
    decodedValue -=
        ((int64_t)1 << (timestampEncodings[index].bitsForValue - 1));
    if (decodedValue >= 0) {
      // [-128,127] becomes [-128,128] without the zero in the middle
      decodedValue++;
    }

    *prevTimestampDelta += decodedValue;
  }

  *prevTimestampValue += *prevTimestampDelta;
  return *prevTimestampValue;
}

double sdsReadNextValue(sds s, uint64_t *bitPos, uint64_t *previousValue, uint8_t* previousLeadingZeros, uint8_t *previousTrailingZeros) {
	uint64_t nonZeroValue = sdsreadbits(s, bitPos, 1);
	if (!nonZeroValue) {
    double* p = (double*)previousValue;
    return *p;
  }

	uint64_t usePreviousBlockInformation = sdsreadbits(s, bitPos, 1);
	uint64_t xorValue;
	if (usePreviousBlockInformation) {
		xorValue = sdsreadbits(s, bitPos, 64 - *previousLeadingZeros - *previousTrailingZeros);
    xorValue <<= *previousTrailingZeros;
	} else {
		uint64_t leadingZeros = sdsreadbits(s, bitPos, kLeadingZerosLengthBits);
    uint64_t blockSize = sdsreadbits(s, bitPos, kBlockSizeLengthBits) + kBlockSizeAdjustment;
		*previousTrailingZeros = 64 - blockSize - leadingZeros;
		xorValue = sdsreadbits(s, bitPos, blockSize);
		xorValue <<= *previousTrailingZeros;
		*previousLeadingZeros = leadingZeros;
	}

	uint64_t value = xorValue ^ *previousValue;
  *previousValue = value;

  double* p = (double*)&value;
  return *p;
}

#ifdef TSDB_TEST_MAIN
#include "tsdb_test.c"
#endif //!TSDB_TEST_MAIN
